package com.uestc.coprocessor.userLog;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;

import com.google.protobuf.ServiceException;
import com.uestc.coprocessor.userLog.UserLogAggregationProtocol.ActiveUserLocationCountService;
import com.uestc.util.JsonUtil;
import com.uestc.util.Log;

public class UserLogAggregationClient {

	public String getActiveUserMap(Table table, String beginDate, String endDate) throws ServiceException, Throwable {
		final UserLogAggregationProtocol.CountRequest request = UserLogAggregationProtocol.CountRequest.newBuilder()
				.setBeginDate(beginDate).setEndDate(endDate).build();
		Map<byte[], Map<String, Long>> resultMap = table.coprocessorService(
				UserLogAggregationProtocol.ActiveUserLocationCountService.class, null, null,
				new Batch.Call<UserLogAggregationProtocol.ActiveUserLocationCountService, Map<String, Long>>() {

					@Override
					public Map<String, Long> call(ActiveUserLocationCountService service) throws IOException {
						// TODO Auto-generated method stub
						BlockingRpcCallback<UserLogAggregationProtocol.CountResponse> rpcCallback = new BlockingRpcCallback<>();
						service.getCounts(null, request, rpcCallback);
						UserLogAggregationProtocol.CountResponse response = rpcCallback.get();
						String result = response.getResult();
						Map<String, Long> map = new HashMap<>();
						// 处理result字符串
						String[] provinceAndValue = result.split(";");
						if (provinceAndValue == null)
							return null;
						for (String pv : provinceAndValue) {
							if (pv == null || pv.isEmpty()) // 判断字符串为空
								continue;
							String[] kv = pv.split(":");
							if (kv.length < 2) // 保证字符串可被分割
								continue;
							map.put(kv[0], Long.valueOf(kv[1]));
						}
						return map;
					}
				});

		Map<String, Long> collection = new HashMap<>();
		for (Map<String, Long> item : resultMap.values()) {
			for (Map.Entry<String, Long> i : item.entrySet()) {
				Long value = collection.get(i.getKey());
				if (value == null) {
					collection.put(i.getKey(), i.getValue());
				} else {
					collection.put(i.getKey(), value + i.getValue());
				}
			}
		}

		// 利用Map生成Json字符串
		String jsonString = JsonUtil.createJsonString(collection);
		Log.logger.info(jsonString);
		return jsonString;
	}
}
